package flink_p1

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration

import java.util.{Timer, TimerTask}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, createTypeInformation}

import scala.collection.mutable
import scala.io.{BufferedSource, Source}

object FlinkTest_22_table2_timer_update_file {

  /**
   * 关联维度表：使用Timer定时更新file
   *
   * @param args
   */
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val socketStream: DataStream[String] = env.socketTextStream("127.0.0.1", 8889)

    socketStream.map(new RichMapFunction[String, (String, String)] {

      val idCityMap = new mutable.HashMap[String, String]()

      override def open(parameters: Configuration): Unit = {

        updateConfigFile()
        val timer = new Timer()
        timer.schedule(new TimerTask {
          override def run(): Unit = {
            updateConfigFile()
          }
        }, 1000, 2000)
        // 1s之后开始执行，每2秒执行一次
      }

      def updateConfigFile() = {
        val source: BufferedSource = Source.fromFile("data/id2city.txt")
        for (elem <- source.getLines()) {
          val arr: Array[String] = elem.split(" ")
          idCityMap.put(arr(0), arr(1))
        }
      }


      override def map(value: String): (String, String) = {
        (value, idCityMap.getOrElse(value, "null"))
      }
    }).print()

    env.execute()
  }

}
